package com.lsx143.wordcount.day3.fight;

import com.lsx143.wordcount.bean.OrderEvent;
import com.lsx143.wordcount.bean.TxEvent;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedCoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;

public class Test4_Order {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<OrderEvent> OSOrder = env
                .readTextFile("input/OrderLog.csv")
                .map(new MapFunction<String, OrderEvent>() {
                    @Override
                    public OrderEvent map(String line) throws Exception {
                        String[] datas = line.split(",");
                        return new OrderEvent(
                                Long.valueOf(datas[0]),
                                datas[1],
                                datas[2],
                                Long.valueOf(datas[3])
                        );
                    }
                });
        SingleOutputStreamOperator<TxEvent> OSTx = env
                .readTextFile("input/ReceiptLog.csv")
                .map(new MapFunction<String, TxEvent>() {
                    @Override
                    public TxEvent map(String line) throws Exception {
                        String[] datas = line.split(",");
                        return new TxEvent(
                                datas[0],
                                datas[1],
                                Long.valueOf(datas[2])
                        );
                    }
                });
        ConnectedStreams<OrderEvent, TxEvent> DSSConnected = OSOrder.connect(OSTx);
        DSSConnected
                .keyBy("txId", "txId")
                .process(new KeyedCoProcessFunction<String, OrderEvent, TxEvent, String>() {
                    Map<String, OrderEvent> orderMap = new HashMap<>();
                    Map<String, TxEvent> txMap = new HashMap<>();

                    @Override
                    public void processElement1(OrderEvent orderEvent, Context ctx, Collector<String> out) {
                        if (txMap.containsKey(orderEvent.getTxId())) {
                            //1、匹配,下单号和付款号匹配
                            out.collect("订单:" + orderEvent.getTxId() + ",对账成功");
                            txMap.remove(orderEvent.getTxId());
                        } else {
                            //2、未匹配，下单了，还没付款
                            orderMap.put(orderEvent.getTxId(), orderEvent);
                            out.collect("订单:" + orderEvent.getTxId() + ",对账【失败】");

                        }
                    }

                    @Override
                    public void processElement2(TxEvent txEvent, Context ctx, Collector<String> out) {
                        if (orderMap.containsKey(txEvent.getTxId())) {
                            //1、匹配,下单号和付款号匹配
                            out.collect("订单:" + txEvent.getTxId() + ",对账成功");
                            orderMap.remove(txEvent.getTxId());
                        } else {
                            //2、未匹配，下单了，还没付款
                            txMap.put(txEvent.getTxId(), txEvent);
                            out.collect("订单:" + txEvent.getTxId() + ",对账【失败】");
                        }
                    }
                }).print();
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}